//! Mailbox interfaces.
//!
//! These interfaces are intended to be persistent and tolerant of intermittent
//! failures.  Ie. if the connection to the remote drops then we should keep
//! reconnecting in the background and trying to fulfill the duty.  Timeouts can
//! be implemented around these interfaces.

use aspect_msgqueue::{MsgRange, QueueIdent, QueueList, QueueNamespace, QueueRef};
use async_trait::async_trait;

use crate::{
    batch::MsgSubmitBatch,
    view::{QueueInboxView, ReceivedMsg},
    MbResult, MsgFetchBatch, QueueOutboxView,
};

/// A remote inbox that we can push messages to.
#[async_trait]
pub trait RemotePushInbox {
    /// Fetches queues in a namespace that can be pushed to.
    async fn query_queues(
        &self,
        ns: QueueNamespace,
        start: QueueIdent,
        limit: u64,
    ) -> MbResult<QueueList>;

    /// Fetches our view of an queue inbox.
    ///
    /// It might change at any time.
    async fn fetch_cur_view(&self, queue: QueueRef) -> MbResult<QueueInboxView>;

    /// Pushes a batch of messages.  If this function returns successfully then
    /// the remote has durably stored the messages and the sender can stop sending them.
    ///
    /// This can include messages that have already been processed by the
    /// receiver.  These should be quietly discarded.
    ///
    /// This is for a "sender active push" relationship.
    async fn push_batch(&self, queue: QueueRef, batch: MsgSubmitBatch) -> MbResult<()>;
}

/// A remote outbox that we can pull messages from.
#[async_trait]
pub trait RemotePullOutbox {
    /// Fetches queues in a namespace that can be pulled from.
    async fn query_queues(
        &self,
        ns: QueueNamespace,
        start: QueueIdent,
        limit: u64,
    ) -> MbResult<QueueList>;

    /// Fetch our view of a queue outbox.
    async fn fetch_cur_view(&self, queue: QueueRef) -> MbResult<QueueOutboxView>;

    /// Updates the read position of a remote queue outbox.
    ///
    /// This allows the remote to discard old messages that are no longer
    /// needed.
    async fn update_read_pos(&self, queue: QueueRef, pos: u64) -> MbResult<()>;

    /// Fetches a batch of messages in a range from a queue.
    ///
    /// This is for a "receiver pull" relationship.
    async fn pull_range(&self, queue: QueueRef, range: MsgRange) -> MbResult<MsgFetchBatch>;
}

/// A remote outbox that we can request messages to be pushed from.
#[async_trait]
pub trait RemoteReqPushOutbox {
    /// Fetches queues in a namespace that we can read from.
    async fn query_queues(
        &self,
        ns: QueueNamespace,
        start: QueueIdent,
        limit: u64,
    ) -> MbResult<QueueList>;

    /// Fetch our view of a queue outbox.
    async fn fetch_cur_view(&self, queue: QueueRef) -> MbResult<QueueOutboxView>;

    /// Updates the read position of a remote queue outbox.
    ///
    /// This allows the remote to discard old messages that are no longer
    /// needed.
    async fn update_read_pos(&self, queue: QueueRef, pos: u64) -> MbResult<()>;

    /// Subscribes to push messages on a queue, returning a handle that can be
    /// read from.  Needs to specify an index that we start reading from.
    ///
    /// This is for a "receiver requested push" relationship.
    async fn subscribe_queue(
        &self,
        queue: QueueRef,
        start_pos: u64,
    ) -> MbResult<QueueSubscriptionHandle>;
}

/// A generic subscription to a queue.
///
/// This is for a "receiver requested push" relationship.
#[async_trait]
pub trait QueueSubscription {
    /// Gets the queue being used here.
    fn queue_ref(&self) -> QueueRef;

    /// Returns the index of the last message received.
    async fn get_last_msg(&self) -> MbResult<u64>;

    /// Receives the next message available.
    ///
    /// Intended to be wrapped as a stream.
    async fn recv_next(&self) -> MbResult<ReceivedMsg>;

    /// Updates the read position of the queue so that the upstream can discard
    /// old messages.
    async fn update_read_pos(&self, pos: u64) -> MbResult<()>;
}

/// Handle to a queue push subscription.
///
/// This exists because we can't have associated types in `async_trait` traits.
pub struct QueueSubscriptionHandle {
    queue_ref: QueueRef,
    inner: Box<dyn QueueSubscription>,
}

impl QueueSubscriptionHandle {
    pub fn new(inner: Box<dyn QueueSubscription>) -> Self {
        let queue_ref = inner.queue_ref();
        Self { queue_ref, inner }
    }

    pub fn queue_ref(&self) -> &QueueRef {
        &self.queue_ref
    }

    pub async fn get_last_msg(&self) -> MbResult<u64> {
        self.inner.get_last_msg().await
    }

    pub async fn recv_next(&self) -> MbResult<ReceivedMsg> {
        self.inner.recv_next().await
    }

    pub async fn update_read_pos(&self, pos: u64) -> MbResult<()> {
        self.inner.update_read_pos(pos).await
    }
}
